package com.demo.mq.kafka.producer;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;

/**
 * @author lee9213@163.com
 * @version 1.0
 * @date 2017-04-04 20:40
 */
@Component
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    public void send(){
        kafkaTemplate.send("my-replicated-topic","xiaojf");
        kafkaTemplate.send("my-replicated-topic","xiaojf");

        kafkaTemplate.metrics();

        kafkaTemplate.execute(new KafkaOperations.ProducerCallback<String, String, Object>() {

            @Override
            public Object doInKafka(Producer<String, String> producer) {

                return null;
            }
        });

        /**
         * 消息发送的监听器，用于回调返回信息
         */
        kafkaTemplate.setProducerListener(new ProducerListener<String, String>() {
            @Override
            public void onSuccess(String s, Integer integer, String s2, String s3, RecordMetadata recordMetadata) {

            }

            @Override
            public void onError(String s, Integer integer, String s2, String s3, Exception e) {

            }

            @Override
            public boolean isInterestedInSuccess() {
                return false;
            }
        });
    }
}
